package service

import (
	"encoding/json"
	"errors"
	"github.com/mohae/deepcopy"
	"gorm.io/gorm"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"
	"vrcm/src/vrcm/app/admin/models"
	"vrcm/src/vrcm/app/admin/models/live_schedule"
	liveDto "vrcm/src/vrcm/app/admin/service/dto/live"
	dto "vrcm/src/vrcm/app/admin/service/dto/live_schedule"
	StreamHubDto "vrcm/src/vrcm/app/admin/service/dto/stream_hub"
	liveService "vrcm/src/vrcm/app/admin/service/live"
	streamHubService "vrcm/src/vrcm/app/admin/service/stream_hub"
	"vrcm/src/vrcm/common/apis"
	commonDto "vrcm/src/vrcm/common/dto"
	"vrcm/src/vrcm/common/service"
	ext "vrcm/src/vrcm/envconfig"
)

type CortexTaskService struct {
	service.Service
}

func (e *CortexTaskService) CreateCortexTasks(wg *sync.WaitGroup, workTotalNum int, partNum *int, cortexTasks []live_schedule.CortexTask, offMap map[int]live_schedule.TrafficOff, upMap map[int]live_schedule.TrafficOn, basic live_schedule.ChannelBasic, sourcesMap map[int]live_schedule.SourceBasic, cortexOutputMap map[int]live_schedule.CortexOutput) error {

	now := time.Now().Unix()
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	liveService.Log = e.Log
	streamHubService.Log = e.Log
	staticsService := &ChannelStatisticsService{}
	staticsService.Orm = e.Orm
	staticsService.Log = e.Log
	var err error
	var mu sync.Mutex
	//存储软矩阵任务
	//taskMap := make(map[string]live_schedule.CortexTask)
	sourceIds := make([]int, 0, 5)

	//组装原始通道配置关系key 和存储特殊处理逻辑
	for _, v := range cortexTasks {
		v := v
		////off := offMap[v.TrafficOffId]
		////存储软矩阵任务
		//originKey := v.TenantId + models.Delimiter + strconv.Itoa(v.ChannelId) + models.Delimiter + strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.MediaProcessId) + models.Delimiter + strconv.Itoa(v.CortexOutputId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
		//taskMap[originKey] = v
		//e.Log.Trace("所有任务：", taskMap)
		sourceIds = append(sourceIds, v.SourceId)
	}

	if sourcesMap == nil {
		//获取所有的信源
		sources := make([]live_schedule.SourceBasic, 0, 5)
		err = e.Orm.Model(live_schedule.SourceBasic{}).Where(" id in ?", sourceIds).Find(&sources).Error
		if err != nil {
			e.Log.Errorf("SaveWorkflow get SourceBasic error:%s \r\n", err)
			//tx.Rollback()
			return err
		}
		sourcesMap = make(map[int]live_schedule.SourceBasic)
		for _, v := range sources {
			v := v
			sourcesMap[v.Id] = v
		}
	}

	e.Log.Trace("信源列表：", sourcesMap)
	//自动调度任务
	//taskCh := make(chan live_schedule.CortexTask, 3)
	//需要启动的任务记录id
	startIds := make([]int, 0, 5)
	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(cortexTasks); i++ {
	//		taskCh <- cortexTasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()
	//
	//if len(taskCh) > 0 {
	//	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Update("channel_state", 4).Error
	//	if err != nil {
	//		return err
	//	}
	//}

	for i := 0; i < len(cortexTasks); i++ {
		wg.Add(1)
		go func(v live_schedule.CortexTask, wg *sync.WaitGroup) {
			defer wg.Done()
			task := v
			pullDomain := ""
			source, ok := sourcesMap[v.SourceId]
			if !ok {
				mu.Lock()
				*partNum++
				mu.Lock()
				e.Log.Trace("信源不存在")
				return
			}

			up := upMap[v.TrafficOnId]
			off := offMap[v.TrafficOffId]
			//如果是srt拉流或者httpts走的还是内网
			if off.TrafficOffType == 3 || (off.TrafficOffType == 4 && off.StreamType == 2) {
				off.Area = 2
			}
			req, autoReq, err := e.GenerateCortexParam(up, off, &task, source, cortexOutputMap)
			if err != nil {
				mu.Lock()
				*partNum++
				mu.Lock()
				e.Log.Errorf("GenerateCortexParam error:%s \r\n", err)
				return
			}

			////如果是转推第三方并且是已用转推任务则需要更新任务
			//if off.StreamType == 1 && len(task.TaskId) > 0 {
			//	req.Id = task.TaskId
			//	target.Id = task.OutNodeId
			//	hub, err = streamHubService.UpdateStreamHub(&req)
			//} else {
			//
			//	hub, err = streamHubService.CreateStreamHub(&req)
			//}

			taskIds := make([]string, 0, 3)
			hub, err := streamHubService.CreateStreamHub(&req)
			if err != nil {
				e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
			}

			if hub != nil {
				if len(hub.Input.Sources) > 0 {
					task.InUrl = hub.Input.Sources[0].UdpAddress
					task.InNodeId = hub.Input.Sources[0].Id
				}

				//如果是外网拉流就不需要再次接收返回的输出url
				if len(hub.Output.Targets) > 0 {
					if !((off.Area == 1 || off.Area == 3) && off.StreamType == 2) {
						task.OutUrl = hub.Output.Targets[0].Url
					}
					if off.IsCustom == 1 {
						task.CustomOutUrl = hub.Output.Targets[0].RedefinePlayUrlParams.Path + "/" + hub.Output.Targets[0].RedefinePlayUrlParams.Sub + ".m3u8"
					}
					task.OutNodeId = hub.Output.Targets[0].Id
				}
				//如果下车点是httpts取AdditionalPublishUrl
				if off.TrafficOffType == 3 {
					task.OutUrl = hub.Output.Targets[0].AdditionalPublishUrl
				}
				task.StateUpdateTime = now
				task.TaskState = 3
				task.IsEnable = 1
				task.TaskId = hub.TaskId
				task.UpdateTime = now
				task.OutputStreamId = hub.OutputStreamName
				task.StateUpdateTime = now
				err = e.Orm.Save(&task).Error
				if err != nil {
					e.Log.Errorf("save CortexTask error:%s \r\n", err)
				}

			}

			hub1 := &StreamHubDto.StreamHubRes{}
			//自动调度内网情况
			if off.Area == 3 && off.StreamType == 2 && (off.TrafficOffType == 1 || off.TrafficOffType == 2 || off.TrafficOffType == 6) {
				//req1.Output = StreamHubDto.Output{
				//	Targets: []StreamHubDto.Target{target1},
				//	Delay:   delay,
				//	Acl:     StreamHubDto.Acl{},
				//}
				//
				//req1.OutputStreamName = pullUrlPart
				hub1, err = streamHubService.CreateStreamHub(&autoReq)
				if err != nil {
					e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
				}
				if hub1 != nil {
					v1 := deepcopy.Copy(task).(live_schedule.CortexTask)
					v1.Id = 0

					if len(hub1.Input.Sources) > 0 {
						v1.InUrl = hub1.Input.Sources[0].UdpAddress
						v1.InNodeId = hub1.Input.Sources[0].Id
					}

					//如果是外网拉流就不需要再次接收返回的输出url
					if len(hub1.Output.Targets) > 0 {
						taskIds = append(taskIds, hub1.Output.Targets[0].Id)
						v1.OutUrl = hub1.Output.Targets[0].Url
						pullDomain = v1.OutUrl[0:strings.LastIndex(v1.OutUrl, "/")]
						v1.OutNodeId = hub1.Output.Targets[0].Id
						if hub1.Output.Targets[0].RedefinePlayUrlEnable == "1" {
							path := hub1.Output.Targets[0].RedefinePlayUrlParams.Path + "/" + hub1.Output.Targets[0].RedefinePlayUrlParams.Sub + ".m3u8"
							v1.CustomOutUrl = path
						}
					}
					//如果下车点是httpts取AdditionalPublishUrl
					if off.TrafficOffType == 3 {
						task.OutUrl = hub1.Output.Targets[0].AdditionalPublishUrl
					}
					task.OutUrl = pullDomain + "/" + task.OutUrl[strings.LastIndex(task.OutUrl, "/")+1:]
					v1.StateUpdateTime = time.Now().Unix()
					v1.TaskState = 3
					v1.TaskId = hub1.TaskId
					v1.Area = 2
					v1.IsEnable = 1
					v1.OutputStreamId = hub1.OutputStreamName
					v1.UpdateTime = time.Now().Unix()
					err = e.Orm.Save(&v1).Error
					if err != nil {
						e.Log.Errorf("save CortexTask error:%s \r\n", err)
					}
				}
			}
			//如果通道状态是启用 就开启协助任务
			e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Find(&basic)
			if basic.ChannelState == 1 {
				startIds = append(startIds, task.Id)

				if (source.Kind == 2 || source.Kind == 3) && len(source.TaskId) > 0 {
					err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: source.TaskId})
					if err != nil {
						e.Log.Errorf("StartStreamHub error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}
				}
				err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: hub.TaskId})
				if err != nil {
					e.Log.Errorf("StartStreamHub error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
				if len(hub1.TaskId) > 0 {
					err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: hub1.TaskId})
					if err != nil {
						e.Log.Errorf("StopStreamHub error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}
					//记录任务启动记录用于统计
					err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, hub1.TaskId, 1, now)
					if err != nil {
						e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
					}
				}

				//创建调度统计记录
				err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, task.TaskId, 1, now)
				if err != nil {
					e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
				}
				param := make(map[string]interface{})
				param["is_enable"] = 1
				param["update_time"] = time.Now().Unix()
				param["state_update_time"] = time.Now().Unix()
				err = e.Orm.Model(live_schedule.CortexTask{}).Where("id = ?", v.Id).Updates(param).Error
				if err != nil {
					e.Log.Errorf("update CortexTask error:%s \r\n", err)
				}
			}
			//设置黑白名单
			channelTaskService := ChannelTaskService{}
			channelTaskService.Log = e.Log
			taskIds = append(taskIds, v.OutNodeId)
			channelTaskService.ProcessBlackWhite(taskIds, off)

			//存储任务历史记录
			if off.StreamType == 2 && len(task.TaskId) > 0 && len(v.OutUrl) > 0 {
				history := live_schedule.ScheduleTaskHistory{}
				history.TenantId = v.TenantId
				history.ChannelId = v.ChannelId
				history.PullUrl = v.OutUrl
				history.OffId = v.TrafficOffId
				history.CreateTime = now
				history.UpdateTime = now
				err = e.Orm.Save(&history).Error
				if err != nil {
					e.Log.Errorf("Save ScheduleTaskHistory error:%s \r\n", err)
				}
			}
			//mu.Lock()
			//*partNum++
			//err = e.CalculateProgress(*partNum, workTotalNum, basic)
			//if err != nil {
			//	e.Log.Trace("CalculateProgress error：", err)
			//}
			//mu.Unlock()
		}(cortexTasks[i], wg)
	}
	return nil
}

func (e *CortexTaskService) DeleteCortexTask(wg *sync.WaitGroup, workTotalNum int, partNum *int, tasks []live_schedule.CortexTask, sourcesMap map[int]live_schedule.SourceBasic, basic live_schedule.ChannelBasic, liveService *liveService.LiveBasic, streamHubService *streamHubService.StreamHubService, liveBasicService SourceBasicService) error {

	//var m sync.Mutex
	liveBasicService.Orm = e.Orm
	// 创建任务通道
	//taskCh := make(chan live_schedule.CortexTask, 3)
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()

	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		func(v live_schedule.CortexTask, wg *sync.WaitGroup) {
			//source := sourcesMap[v.SourceId]
			////删除自定义
			//if source.Kind == 2 && len(source.TaskId) > 0 {
			//	err := liveBasicService.DeleteCustomSource(source)
			//	if err != nil {
			//		e.Log.Errorf("DeleteCustomSource error:%s \r\n", err)
			//	}
			//}
			defer wg.Done()
			//删除协转任务流
			if len(v.TaskId) > 0 {
				err := streamHubService.RemoveStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("RemoveStreamHub error:%s \r\n", err)
				}
				param := make(map[string]interface{})
				param["end_time"] = now
				err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0", v.TaskId).Updates(&param).Error
				if err != nil {
					e.Log.Errorf("updateScheduleRecord error:%s \r\n", err)
				}
			}

			if len(v.LiveId) > 0 {
				err := liveService.RemoveLive(&liveDto.StopLiveReq{TenantId: v.TenantId, LiveId: v.LiveId})
				if err != nil {
					e.Log.Errorf("RemoveLive error:%s \r\n", err)
				}
			}
			err := e.Orm.Delete(&v).Error
			if err != nil {
				e.Log.Errorf("SaveWorkflow_Delete error:%s \r\n", err)
			}
			//m.Lock()
			//*partNum++
			//err = e.CalculateProgress(*partNum, workTotalNum, basic)
			//if err != nil {
			//	e.Log.Trace("CalculateProgress error：", err)
			//}
			//m.Unlock()
		}(tasks[i], wg)
	}
	return nil

}

func (e *CortexTaskService) StartCortexTasks(wg *sync.WaitGroup, tasks []live_schedule.CortexTask, workNum int, partNum *int, param map[string]interface{}, sourceMap map[int]live_schedule.SourceBasic) error {
	// 创建任务通道
	//taskCh := make(chan live_schedule.CortexTask, 3)
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	sourceBasicService := &SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	staticsService := &ChannelStatisticsService{}
	staticsService.Orm = e.Orm
	staticsService.Log = e.Log
	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()
	//var m sync.Mutex
	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		func(v live_schedule.CortexTask, wg *sync.WaitGroup) {
			//up := sourceMap[v.SourceId]
			//if up.Kind == 2 {
			//	err := sourceBasicService.StartSource(&dto.StartSourceReq{TenantId: v.TenantId, TaskId: up.TaskId, Id: v.SourceId})
			//	if err != nil {
			//		e.Log.Errorf("StartSource error:%s \r\n", err)
			//	}
			//}
			defer wg.Done()
			if len(v.TaskId) > 0 && v.IsEnable == 1 {
				//开启协助任务
				err := streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("StartStreamHub error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
				//创建调度统计记录
				err = staticsService.CreateScheduleRecord(v.TenantId, v.ChannelId, v.TaskId, 1, now)
				if err != nil {
					e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
				}
			}
			//m.Lock()
			//*partNum++
			//result := float64(*partNum) / float64(workNum)
			//formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
			//if workNum == *partNum {
			//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Updates(param).Error
			//	if err != nil {
			//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
			//	}
			//} else {
			//	param["progress"] = formattedResult
			//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Update("progress", formattedResult).Error
			//	if err != nil {
			//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
			//	}
			//}
			//
			//e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
			//m.Unlock()
		}(tasks[i], wg)
	}

	return nil

}

func (e *CortexTaskService) StopCortexTasks(wg *sync.WaitGroup, tasks []live_schedule.CortexTask, workNum int, partNum *int, param map[string]interface{}, sourceMap map[int]live_schedule.SourceBasic) error {
	// 创建任务通道
	//taskCh := make(chan live_schedule.CortexTask, 3)
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	sourceBasicService := &SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()

	//var mu sync.Mutex
	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		func(v live_schedule.CortexTask, wg *sync.WaitGroup) {
			//up := sourceMap[v.SourceId]
			//if up.Kind == 2 {
			//	err := sourceBasicService.StopSource(&dto.StopSourceReq{TenantId: v.TenantId, TaskId: up.TaskId})
			//	if err != nil {
			//		e.Log.Errorf("StopSource error:%s \r\n", err)
			//	}
			//}
			defer wg.Done()
			if len(v.TaskId) > 0 && v.IsEnable == 1 {
				err := streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{Id: v.TaskId, TenantId: v.TenantId})
				if err != nil {
					e.Log.Errorf("StopStreamHub error:%s \r\n", err)
				}
				//更新通道任务状态为离线
				param1 := make(map[string]interface{})
				param1["update_time"] = now
				param1["task_state"] = 3
				err = e.Orm.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id=? and task_id=?", v.TenantId, v.ChannelId, v.TaskId).Updates(&param1).Error
				if err != nil {
					e.Log.Errorf("StopCortexTasks error:%s \r\n", err)
				}
			}

			//更新任务记录结束时间
			param2 := make(map[string]interface{})
			param2["end_time"] = now
			err := e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =1", v.TaskId).Updates(&param2).Error
			if err != nil {
				e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
			}

			//mu.Lock()
			//*partNum++
			//result := float64(*partNum) / float64(workNum)
			//formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
			//if *partNum == workNum {
			//	param["progress"] = formattedResult
			//	err = e.Orm.Model(live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Updates(param).Error
			//	if err != nil {
			//		e.Log.Errorf("Update ChannelTask error:%s \r\n", err)
			//	}
			//} else {
			//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Update("progress", formattedResult).Error
			//	if err != nil {
			//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
			//	}
			//}
			//e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
			//mu.Unlock()
		}(tasks[i], wg)
	}

	return nil

}

func (e *CortexTaskService) UpdateCortexOutput(c *dto.UpdateCortexOutput) error {

	streamHubService := streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	streamHubService.Orm = e.Orm

	out := live_schedule.CortexOutput{}
	err := e.Orm.Model(&live_schedule.CortexOutput{}).Where("out_name=?", c.OutputName, c.Id).Find(&out).Error
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("get CortexOutput error:%s \r\n", err)
		return err
	}
	if out.Id > 0 && out.Id != c.Id {
		return errors.New(string(apis.OutPutNameRepeatError.Code))
	}

	err = e.Orm.Model(live_schedule.CortexOutput{}).Where("id=?", c.Id).Update("out_name", c.OutputName).Error
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("UpdateCortexOutput error:%s \r\n", err)
		return err
	}

	//获取所有任务
	tasks := make([]live_schedule.CortexTask, 0, 5)
	err = e.Orm.Model(live_schedule.CortexTask{}).Where("cortex_output_id=?", c.Id).Find(&tasks).Error
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		return err
	}

	//当前正在输出的上车点
	upId := 0

	tenant_id := ""
	channel_id := 0
	if len(c.CortexInfos) > 0 {

		pushInfoMap := make(map[int]string)
		pullInfoMap := make(map[int]string)
		for _, v := range c.CortexInfos {
			//key := upId + models.Delimiter + strconv.Itoa(v.OffId) + models.Delimiter + strconv.Itoa(c.Id)
			//转推地址
			pushInfoMap[v.OffId] = v.PushUrl
			//自定义地址修改操作
			pullInfoMap[v.OffId] = v.PullUrl
		}
		//有输出的矩阵
		useCortexOutputIdMap := make(map[int]struct{})
		cortexOutputIds := make([]string, 0, 5)

		now := time.Now().Unix()
		//储存正在输出的输出点，新添加的转推地址会自动生成转推任务
		for _, v := range tasks {
			v := v
			if len(v.TaskId) > 0 {
				useCortexOutputIdMap[v.CortexOutputId] = struct{}{}
				upId = v.TrafficOnId
			}
		}
		outputMap := make(map[string]struct{})
		for k, v := range tasks {
			v := v
			pushUrl, _ := pushInfoMap[v.TrafficOffId]
			pullInfo, ok2 := pullInfoMap[v.TrafficOffId]
			//过滤掉不是当前输出的山车点
			if v.TrafficOnId != upId && upId > 0 {
				*(&tasks[k].PushUrl) = pushUrl
				*(&tasks[k].CustomOutUrl) = pullInfo
				continue
			}
			key := strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.CortexOutputId)
			tenant_id = v.TenantId
			channel_id = v.ChannelId

			//编辑推流地址需要删除之前的推流任务
			if len(v.TaskId) > 0 && len(pushUrl) > 0 && pushUrl != v.PushUrl {
				err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					//tx.Rollback()
					e.Log.Errorf("StopStreamHub error:%s \r\n", err)
					return err
				}
				err = streamHubService.DeleteStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					//tx.Rollback()
					e.Log.Errorf("DeleteStreamHub error:%s \r\n", err)
					return err
				}
				param := make(map[string]interface{})
				param["end_time"] = now
				err := e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0", v.TaskId).Updates(param).Error
				if err != nil {
					e.Log.Errorf("updateScheduleRecord error:%s \r\n", err)
					return err
				}
				*(&tasks[k].TaskId) = ""
				v.TaskId = ""

			}

			e.Log.Trace("转推地址：", pushUrl, "&&", v.TrafficOnId)
			*(&tasks[k].PushUrl) = pushUrl
			//输出点有输出任务且对应下车点没有生成任务需要保存创建的任务
			if len(v.TaskId) <= 0 && (len(pushUrl) > 0 || len(pullInfo) > 0) {
				_, ok := useCortexOutputIdMap[v.CortexOutputId]
				if ok {
					_, ok1 := outputMap[key]
					if !ok1 {
						cortexOutputIds = append(cortexOutputIds, key)
					}
					outputMap[key] = struct{}{}
				}
			}

			//编辑自定义拉流地址
			if ok2 && len(pullInfo) > 0 && len(v.TaskId) > 0 && pullInfo != v.CustomOutUrl {
				arr := strings.Split(pullInfo, "/")
				if len(arr) > 1 {
					index := strings.Index(arr[1], ".")
					sub := arr[1][0:index]
					//如果编辑自定义参数且任务是启用状态需要调用编辑自定义参数接口 && channelBasic.ChannelState != 2
					if len(v.TaskId) > 0 {
						err = streamHubService.EditRefinePlayUrl(&StreamHubDto.EditRefinePlayUrlReq{TenantId: v.TenantId, Id: v.OutNodeId, RedefinePlayUrlEnable: "1", RedefinePlayUrlParams: StreamHubDto.RedefinePlayUrlParams{Path: arr[0], Sub: sub}})
						if err != nil {
							//tx.Rollback()
							e.Log.Errorf("EditRefinePlayUrl error:%s \r\n", err)
							return err
						}
					}
				}
			}
			*(&tasks[k].CustomOutUrl) = pullInfo
		}

		if len(tasks) > 0 {
			err = e.Orm.Model(live_schedule.CortexTask{}).Save(&tasks).Error
			if err != nil {
				e.Log.Errorf("save CortexTask error:%s \r\n", err)
				return err
			}
		}

		for _, v := range cortexOutputIds {
			arr := strings.Split(v, models.Delimiter)
			if len(arr) < 2 {
				//tx.Rollback()
				e.Log.Errorf("Split cortexOutputIds error:%s \r\n", err)
				return err
			}
			upId, _ := strconv.Atoi(arr[0])
			outputId, _ := strconv.Atoi(arr[1])
			err = e.CreateCortexTask(nil, &dto.CreateCortexTaskReq{TenantId: tenant_id, ChannelId: channel_id, UpId: upId, CortexOutputId: outputId})
			if err != nil {
				//tx.Rollback()
				e.Log.Errorf("create CortexTask error:%s \r\n", err)
				return err
			}
		}
	}
	return nil
}

func (e *CortexTaskService) SaveCortex(c *dto.SaveCortexReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("Service SaveCortex error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	streamHubService := streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	streamHubService.Orm = e.Orm
	//获取调度基本信息
	var basic live_schedule.ChannelBasic
	err := tx.Model(live_schedule.ChannelBasic{}).Where("id =? ", c.ChannelId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		return err
	}
	//获取矩阵变化
	var mediaProcess live_schedule.MediaProcess
	err = tx.Model(live_schedule.MediaProcess{}).Where("channel_id=? and media_type=2 ", c.ChannelId).Find(&mediaProcess).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return err
	}

	if mediaProcess.Id <= 0 {
		return errors.New("没有找到软矩阵")
	}

	//根据频道获取上车点
	upIds := make([]int, 0, 5)
	err = tx.Model(live_schedule.UpMedia{}).Select("up_id").Where("channel_id=? and media_id=?", c.ChannelId, mediaProcess.Id).Find(&upIds).Error
	if err != nil {
		e.Log.Errorf("get UpMedia error:%s \r\n", err)
		return err
	}

	//根据频道获取上车点
	ups := make([]live_schedule.TrafficOn, 0, 5)
	err = tx.Model(live_schedule.TrafficOn{}).Where("id in ?", upIds).Find(&ups).Error
	if err != nil {
		e.Log.Errorf("get TrafficOn error:%s \r\n", err)
		return err
	}
	upMap := make(map[int]live_schedule.TrafficOn)
	sourcesIds := make([]int, 0, 3)
	for _, v := range ups {
		v := v
		upMap[v.Id] = v
		sourcesIds = append(sourcesIds, v.SourceId)
	}

	//获取下车点
	OffIds := make([]int, 0, 5)
	err = tx.Model(live_schedule.MediaOff{}).Select("off_id").Where("channel_id=? and media_id=?", c.ChannelId, mediaProcess.Id).Find(&OffIds).Error
	if err != nil {
		e.Log.Errorf("get MediaOff error:%s \r\n", err)
		return err
	}

	offs := make([]live_schedule.TrafficOff, 0, 5)
	err = tx.Model(live_schedule.TrafficOff{}).Where("id in ?", OffIds).Find(&offs).Error
	if err != nil {
		e.Log.Errorf("get TrafficOff error:%s \r\n", err)
		return err
	}

	offMap := make(map[int]live_schedule.TrafficOff)
	for _, off := range offs {
		off := off
		offMap[off.Id] = off
	}
	//获取所有的信源
	sources := make([]live_schedule.SourceBasic, 0, 5)
	err = e.Orm.Model(live_schedule.SourceBasic{}).Where(" id in ?", sourcesIds).Find(&sources).Error
	if err != nil {
		e.Log.Errorf("SaveWorkflow get SourceBasic error:%s \r\n", err)
		return err
	}
	sourceMap := make(map[int]live_schedule.SourceBasic)
	for _, v := range sources {
		v := v
		sourceMap[v.Id] = v
	}

	//转推地址
	pushInfoMap := make(map[int][]dto.PushInfo)
	pullInfoMap := make(map[int][]dto.PullInfo)
	OutputNames := make(map[string]struct{})
	CortexPushInfoMap := make(map[string]map[int][]dto.PushInfo)
	CortexPullInfoMap := make(map[string]map[int][]dto.PullInfo)
	for _, v := range c.CortexInfos {
		v := v
		OutputNames[v.OutputName] = struct{}{}
		if v.OffId <= 0 {
			continue
		}
		off := offMap[v.OffId]
		pushInfos := make([]dto.PushInfo, 0, 5)
		if len(off.PushInfos) > 0 {
			err = json.Unmarshal([]byte(off.PushInfos), &pushInfos)
			if err != nil {
				e.Log.Errorf("Unmarshal PushInfos error:%s \r\n", err)
				return err
			}
			pushInfo := dto.PushInfo{}
			pushInfo.MediaId = mediaProcess.Id
			pushInfo.OffId = off.Id
			pushInfo.CortexOutName = v.OutputName
			pushInfos = append(pushInfos, pushInfo)
			pushInfoMap[v.OffId] = pushInfos
			CortexPushInfoMap[v.OutputName] = pushInfoMap
		}
		pullInfos := make([]dto.PullInfo, 0, 5)
		if len(off.PullInfos) > 0 {
			err = json.Unmarshal([]byte(off.PullInfos), &pullInfos)
			if err != nil {
				e.Log.Errorf("Unmarshal PullInfos error:%s \r\n", err)
				return err
			}
			pllInfo := dto.PullInfo{}
			pllInfo.MediaId = mediaProcess.Id
			pllInfo.OffId = off.Id
			pllInfo.CortexOutName = v.OutputName
			pullInfos = append(pullInfos, pllInfo)
			pullInfoMap[v.OffId] = pullInfos
			CortexPullInfoMap[v.OutputName] = pullInfoMap
		}

	}

	//保存棋盘格输出名
	now := time.Now().Unix()
	cortexOutputs := make([]live_schedule.CortexOutput, 0, 5)
	names := make([]string, 0, 3)
	for k, _ := range OutputNames {
		k := k
		cortexOutput := live_schedule.CortexOutput{}
		cortexOutput.TenantId = c.TenantId
		cortexOutput.MediaProcessId = mediaProcess.Id
		cortexOutput.ChannelId = c.ChannelId
		cortexOutput.OutName = k
		cortexOutput.CreateTime = now
		cortexOutput.UpdateTime = now
		names = append(names, k)
		cortexOutputs = append(cortexOutputs, cortexOutput)
	}

	tempCortexOutputs := make([]live_schedule.CortexOutput, 0, 5)
	err = e.Orm.Model(&live_schedule.CortexOutput{}).Where("out_name in  ? and channel_id = ?", names, c.ChannelId).Find(&tempCortexOutputs).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get CortexOutput error:%s \r\n", err)
		return err
	}

	if len(tempCortexOutputs) > 0 {
		tx.Rollback()
		return errors.New(string(apis.OutPutNameRepeatError.Code))
	}

	err = tx.Model(live_schedule.CortexOutput{}).Create(&cortexOutputs).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("SaveCortexOutPut error:%s \r\n", err)
		return err
	}

	saveOffs := make([]live_schedule.TrafficOff, 0, 3)
	for _, output := range cortexOutputs {
		pullInfoMap, ok := CortexPullInfoMap[output.OutName]
		if ok {
			for offId, infos := range pullInfoMap {
				off, ok := offMap[offId]
				if !ok {
					continue
				}
				for k, i := range infos {
					if i.CortexOutName != output.OutName {
						continue
					}
					*(&infos[k].CortexOutId) = output.Id
				}
				info, _ := json.Marshal(infos)
				off.PullInfos = string(info)
				saveOffs = append(saveOffs, off)
			}
		}

		pushInfoMap, ok := CortexPushInfoMap[output.OutName]
		if ok {
			for offId, infos := range pushInfoMap {
				off, ok := offMap[offId]
				if !ok {
					continue
				}
				for k, i := range infos {
					if i.CortexOutName != output.OutName {
						continue
					}
					*(&infos[k].CortexOutId) = output.Id
				}
				info, _ := json.Marshal(infos)
				off.PushInfos = string(info)
				saveOffs = append(saveOffs, off)
			}
		}
	}

	outs := make([]live_schedule.CortexOutput, 0, 3)
	err = tx.Model(&live_schedule.CortexOutput{}).Where("channel_id=?", c.ChannelId).Find(&outs).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get CortexOutput error:%s \r\n", err)
		return err
	}
	//声称切换任务
	switchMap := make(map[string]string)
	for _, v := range outs {
		v := v
		//如果是软矩阵并且支持vdcm则需要将信源加入vdcm
		if ext.ExtConfig.ServiceConfig.EnableSwitch == 1 && mediaProcess.EnableSwitch == 1 {
			e.Log.Trace("进入切换模式")
			switchReq := &StreamHubDto.SwitchSourceReq{}
			switchReq.Name = ""
			switchReq.TenantId = c.TenantId
			switchReq.Id = v.SwitchTaskId
			switchInput := StreamHubDto.SwitchInput{}
			switchInput.Type = "push"
			switchInput.Format = "udp"
			switchInput.Active = 0
			switchInput.Policy = StreamHubDto.Policy{Mode: "manual"}
			switchInput.Sources = make([]StreamHubDto.Source, 0, 3)
			switchReq.Type = "live"
			switchReq.AutoSwitch = 0

			//原始流切换
			for _, u := range ups {
				u := u
				switchSource := StreamHubDto.Source{}
				source := sourceMap[u.SourceId]
				if u.StreamType == 1 {
					switchSource.Url = source.OutUdpUrl
				} else {
					switchSource.Url = u.TransUdpUrl
				}
				switchSource.Id = u.SwitchId
				switchInput.Sources = append(switchInput.Sources, switchSource)
			}
			switchReq.SwitchInput = switchInput
			switchRes := &StreamHubDto.SwitchSourceRes{}
			if len(v.SwitchTaskId) <= 0 {
				switchRes, err = streamHubService.CreateSwitchSourceTask(switchReq)
			} else {
				switchRes, err = streamHubService.UpdateSwitchSourceTask(switchReq)
			}
			if err != nil {
				e.Log.Errorf("CreateSwitchSourceTask error:%s \r\n", err)
				return err
			}
			if switchRes != nil {
				params := make(map[string]interface{})
				params["switch_task_id"] = switchRes.TaskId
				params["out_udp_url"] = switchRes.OutputAddress
				err = tx.Model(live_schedule.CortexOutput{}).Where("id=?", v.Id).Updates(&params).Error
				if err != nil {
					e.Log.Errorf("update MediaProcess error:%s \r\n", err)
					tx.Rollback()
					return err
				}
				//保存切换任务的主备辅id
				for _, in := range switchRes.Input.Sources {
					in := in
					for _, s := range ups {
						source := sourceMap[s.SourceId]
						if (in.Url == source.OutUdpUrl && s.StreamType == 1) || (in.Url == source.TransOutUdpUrl && s.StreamType == 2) {
							key := strconv.Itoa(s.Id) + models.Delimiter + strconv.Itoa(v.Id)
							switchMap[key] = in.Id
						}
					}
				}
			}
			e.Log.Trace("获取的原始流切换地址：", switchRes.OutputAddress)
		}
	}

	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	//初始化基础数据
	for _, u := range upIds {
		for _, output := range cortexOutputs {
			for _, offId := range OffIds {
				//key := strconv.Itoa(u) + models.Delimiter + strconv.Itoa(offId)
				//创建棋盘格任务
				cortexTask := live_schedule.CortexTask{}
				cortexTask.TenantId = c.TenantId
				cortexTask.ChannelId = c.ChannelId
				cortexTask.SourceId = upMap[u].SourceId
				cortexTask.TrafficOnId = u
				cortexTask.MediaProcessId = mediaProcess.Id
				cortexTask.TrafficOffId = offId
				cortexTask.CortexOutputId = output.Id
				cortexTask.TaskState = 3
				cortexTask.CreateTime = now
				cortexTask.UpdateTime = now
				cortexTask.IsEnable = 2
				//cortexTask.PushUrl = pushInfoMap[key]
				//cortexTask.CustomOutUrl = pullInfoMap[key]
				switchKey := strconv.Itoa(u) + models.Delimiter + strconv.Itoa(output.Id)
				switchId, ok := switchMap[switchKey]
				if ok {
					cortexTask.SwitchNodeId = switchId
				}
				cortexTasks = append(cortexTasks, cortexTask)

			}
		}
	}
	if len(cortexTasks) > 0 {
		err = tx.Model(live_schedule.CortexTask{}).Create(&cortexTasks).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("SaveCortexTask error:%s \r\n", err)
			return err
		}
	}
	if len(saveOffs) > 0 {
		err = tx.Save(&saveOffs).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("saveOffs error:%s \r\n", err)
			return err
		}
	}

	tx.Commit()
	return nil
}

func (e *CortexTaskService) CreateCortexTask(tx *gorm.DB, c *dto.CreateCortexTaskReq) error {

	var wg sync.WaitGroup
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	liveService.Log = e.Log
	streamHubService.Log = e.Log
	//now := time.Now().Unix()
	schedules := make([]live_schedule.ScheduleRecord, 0, 5)
	workTotalNum := 0
	var partNum int
	//获取调度基本信息
	var basic live_schedule.ChannelBasic
	err := e.Orm.Model(live_schedule.ChannelBasic{}).Where("id =? ", c.ChannelId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		return err
	}

	//获取矩阵变化
	var mediaProcess live_schedule.MediaProcess
	err = e.Orm.Model(live_schedule.MediaProcess{}).Where("channel_id=? and media_type=2 ", c.ChannelId).Find(&mediaProcess).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return err
	}
	//获取上车点
	var trafficOn live_schedule.TrafficOn
	err = e.Orm.Model(live_schedule.TrafficOn{}).Where("id=?", c.UpId).Find(&trafficOn).Error
	if err != nil {
		e.Log.Errorf("get TrafficOn error:%s \r\n", err)
		//tx.Rollback()
		return err
	}
	upMap := make(map[int]live_schedule.TrafficOn)
	upMap[trafficOn.Id] = trafficOn
	//获取所有下车点信息
	trafficOffs := make([]live_schedule.TrafficOff, 0, 5)
	offIds := make([]int, 0, 5)
	err = e.Orm.Model(live_schedule.MediaOff{}).Select("off_id").Where("channel_id=?", c.ChannelId).Find(&offIds).Error
	if err != nil {
		e.Log.Errorf("get MediaOff error:%s \r\n", err)
		return err
	}
	err = e.Orm.Model(live_schedule.TrafficOff{}).Where("channel_id=? and id in ?", c.ChannelId, offIds).Find(&trafficOffs).Error
	if err != nil {
		e.Log.Errorf("get TrafficOff error:%s \r\n", err)
		return err
	}
	offMap := make(map[int]live_schedule.TrafficOff)
	for _, v := range trafficOffs {
		v := v
		offMap[v.Id] = v
		offIds = append(offIds, v.Id)
	}

	//获取已经存在的配置
	model, err := e.GetCortexOutInfo(nil, &dto.GetCortexOutInfo{TenantId: c.TenantId, CortexOutputId: c.CortexOutputId, ChannelId: c.ChannelId, UpId: c.UpId})
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("GetCortexOutInfo error:%s \r\n", err)
		return err
	}
	offUrlMap := make(map[string]live_schedule.CortexInfo)
	for _, v := range model.CortexInfos {
		v := v
		key := strconv.Itoa(v.CortexOutId) + models.Delimiter + strconv.Itoa(v.OffId)
		offUrlMap[key] = v
	}

	//获取信源
	var source live_schedule.SourceBasic
	err = e.Orm.Model(live_schedule.SourceBasic{}).Where("id = ?", trafficOn.SourceId).Find(&source).Error
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		return err
	}
	sourceMap := make(map[int]live_schedule.SourceBasic)
	sourceMap[source.Id] = source

	//获取原始任务
	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	err = e.Orm.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id =? and traffic_on_id=? and cortex_output_id=? and media_process_id=? and traffic_off_id in ?", c.TenantId, c.ChannelId, c.UpId, c.CortexOutputId, mediaProcess.Id, offIds).Find(&cortexTasks).Error
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		return err
	}

	for k, v := range cortexTasks {
		v := v
		key := strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
		m, ok := offUrlMap[key]
		if ok {
			*(&cortexTasks[k].PushUrl) = m.PushUrl
			*(&cortexTasks[k].CustomOutUrl) = m.PullUrl
			*(&cortexTasks[k].OutUrl) = m.OutUrl
		}
	}

	//获取输出信息
	var output live_schedule.CortexOutput
	err = e.Orm.Model(live_schedule.CortexOutput{}).Where("id = ?", c.CortexOutputId).Find(&output).Error
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("get CortexOutput error:%s \r\n", err)
		return err
	}
	outputMap := make(map[int]live_schedule.CortexOutput)
	outputMap[output.Id] = output

	//存储软矩阵任务
	taskMap := make(map[string]live_schedule.CortexTask)
	//组装原始通道配置关系key 和存储特殊处理逻辑
	for _, v := range cortexTasks {
		v := v
		//off := offMap[v.TrafficOffId]
		//存储软矩阵任务
		originKey := v.TenantId + models.Delimiter + strconv.Itoa(v.ChannelId) + models.Delimiter + strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.MediaProcessId) + models.Delimiter + strconv.Itoa(v.CortexOutputId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
		taskMap[originKey] = v
		e.Log.Trace("所有任务：", taskMap)
	}

	needCreateCortexTasks := make([]live_schedule.CortexTask, 0, 5)
	for _, off := range offMap {
		//如果是srt拉流或者httpts走的还是内网
		if off.TrafficOffType == 3 || (off.TrafficOffType == 4 && off.StreamType == 2) {
			off.Area = 2
		}
		for _, up := range upMap {
			for _, output := range outputMap {
				originKey := basic.TenantId + models.Delimiter + strconv.Itoa(basic.Id) + models.Delimiter + strconv.Itoa(up.Id) + models.Delimiter + strconv.Itoa(mediaProcess.Id) + models.Delimiter + strconv.Itoa(output.Id) + models.Delimiter + strconv.Itoa(off.Id)
				e.Log.Trace("匹配到的任务：", originKey)
				task, _ := taskMap[originKey]
				if len(task.PushUrl) <= 0 && off.StreamType == 1 || (off.IsCustom == 1 && len(task.CustomOutUrl) <= 0) {
					return errors.New("有未填写的输出地址")
				}

				_, ok := sourceMap[up.SourceId]
				if !ok {
					return errors.New("信源已被删除")
				}
				e.Log.Trace("匹配到的任务：", originKey)
				task, ok1 := taskMap[originKey]
				if !ok1 {
					e.Log.Trace("未找到任务：", originKey)
					continue
				}
				//已经有的输出节点跳过
				if len(task.TaskId) > 0 {
					continue
				}
				workTotalNum++
				needCreateCortexTasks = append(needCreateCortexTasks, task)
			}
		}
	}
	err = e.CreateCortexTasks(&wg, workTotalNum, &partNum, cortexTasks, offMap, upMap, basic, nil, outputMap)
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("CreateCortexTasks error:%s \r\n", err)
		return err
	}
	//新增调度任务记录
	if len(schedules) > 0 {
		err := e.Orm.Model(live_schedule.ScheduleRecord{}).Create(&schedules).Error
		if err != nil {
			e.Log.Errorf(" insert ScheduleRecord error:%s \r\n", err)
			//tx.Rollback()
			return err
		}
	}
	wg.Wait()
	//tx.Commit()
	return nil
}

func (e *CortexTaskService) GenerateCortexParam(up live_schedule.TrafficOn, off live_schedule.TrafficOff, task *live_schedule.CortexTask, source live_schedule.SourceBasic, cortexOutputMap map[int]live_schedule.CortexOutput) (StreamHubDto.StreamHubReq, StreamHubDto.StreamHubReq, error) {
	pullUrlPart := ""
	liveService := &liveService.LiveBasic{}
	liveService.Log = e.Log

	commonService := &CommonService{}
	commonService.Log = e.Log
	//输出参数
	target := StreamHubDto.Target{}
	target.Format = protocols[off.TrafficOffType]
	//输入参数
	sourceParam := StreamHubDto.Source{}
	req := StreamHubDto.StreamHubReq{}

	out := cortexOutputMap[task.CortexOutputId]
	//使用转码源还是原始源 如果是切换任务取切换地址
	if len(out.SwitchTaskId) > 0 {
		sourceParam.Id = task.InNodeId
		sourceParam.Url = out.OutUdpUrl
	} else {
		if up.StreamType == 1 {
			sourceParam.Id = task.InNodeId
			sourceParam.Url = source.OutUdpUrl
		} else {
			sourceParam.Id = task.InNodeId
			sourceParam.Url = up.TransUdpUrl
		}
	}

	if (len(task.PushUrl) <= 0 && off.StreamType == 1) || (off.IsCustom == 1 && len(task.CustomOutUrl) <= 0) {
		e.Log.Trace("有未填写的输出地址")
		return req, req, errors.New("有未填写的输出地址")
	}

	req.Name = ""
	req.Type = "live"
	req.TenantId = task.TenantId
	req.Description = ""
	req.Input = StreamHubDto.Input{
		Format: protocols[5],
		Active: 0,
		Policy: StreamHubDto.Policy{
			Mode: "manual",
		},
		Type:    "push",
		Sources: []StreamHubDto.Source{sourceParam},
	}

	//第三方拉流+内网 输出类型为pull 其他情况为push
	if off.StreamType == 2 && off.Area == 2 {
		target.Type = "pull"
	} else {
		target.Type = "push"
	}

	delay := StreamHubDto.Delay{
		Enable:   "no",
		Duration: 10,
	}
	req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/channel-task/cortex-target-call-back"
	target1 := deepcopy.Copy(target).(StreamHubDto.Target)
	target1.Type = "pull"
	//外网+第三方拉流需要先调用全球直播
	if (off.Area == 3 || off.Area == 1) && off.StreamType == 2 {
		target.Format = "rtmp"
		streamReq := e.GenerateInsertLiveParam(task.TenantId)
		res, err := liveService.CreateLiveStream(streamReq)
		if err != nil {
			return req, req, err
		}

		task.LiveId = res.LiveId
		//记录额外信息
		extraMap := make(map[string]string)
		extraMap["rtmp"] = res.RtmpUrl
		extraMap["hls"] = res.HlsUrl
		extraMap["flv"] = res.FlvUrl

		data, err := json.Marshal(extraMap)
		if err != nil {
			e.Log.Errorf("Marshal error:%s \r\n", err)
		}
		task.ExtraInfo = string(data)
		pushes := strings.Split(res.PushUrl, models.Delimiter)
		if len(pushes) > 0 {
			target.Url = pushes[0]
			task.PushUrl = pushes[0]
		}

		if off.TrafficOffType == 9 {
			off.TrafficOffType = 1
		}
		task.OutUrl = extraMap[protocols[off.TrafficOffType]]
		//如果是自动调度 需要获取外网的sub和内网的域名
		u, err := url.Parse(task.OutUrl)
		if err != nil {
			return req, req, err
		}
		path := strings.TrimPrefix(u.Path, "/")
		if len(path) > 1 {
			streamName := strings.Split(path, "/")[1]
			queryString := strings.Split(streamName, ".")[0]
			pullUrlPart = queryString
			e.Log.Trace("流名：", pullUrlPart)
		}

		//如果是外网并且拉流地址
		if off.Area == 3 && off.StreamType == 2 {
			task.Area = 1
		}

	} else if off.StreamType == 1 {
		target.Url = task.PushUrl
	}

	target.RedefinePlayUrlEnable = "2"
	params := StreamHubDto.RedefinePlayUrlParams{}
	target.RedefinePlayUrlParams = params
	autoReq := deepcopy.Copy(req).(StreamHubDto.StreamHubReq)

	//如果是自定义拉流地址
	if off.StreamType == 2 && off.TrafficOffType == 2 && (off.Area == 2 || off.Area == 3) && off.IsCustom == 1 {
		arr := strings.Split(task.CustomOutUrl, "/")
		if off.Area == 3 {
			if len(arr) > 1 {
				params.Path = arr[0]
				index := strings.Index(arr[1], ".")
				params.Sub = arr[1][0:index]
				target1.RedefinePlayUrlParams = params
			}
		} else {
			if len(arr) > 1 {
				target.RedefinePlayUrlEnable = "1"
				params.Path = arr[0]
				index := strings.Index(arr[1], ".")
				params.Sub = arr[1][0:index]
				target.RedefinePlayUrlParams = params
			}
		}

	}
	//创建协助任务
	req.Output = StreamHubDto.Output{
		Targets: []StreamHubDto.Target{target},
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}
	autoReq.Output = StreamHubDto.Output{
		Targets: []StreamHubDto.Target{target1},
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}
	autoReq.OutputStreamName = pullUrlPart
	return req, autoReq, nil
}

func (e *CortexTaskService) DeleteCortexTaskByOutputName(c *dto.DeleteCortexTaskReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("Service SaveCortex error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	liveService.Log = e.Log
	streamHubService.Log = e.Log

	//获取矩阵变化
	var mediaProcess live_schedule.MediaProcess
	err := tx.Model(live_schedule.MediaProcess{}).Where("channel_id=? and media_type=2 ", c.ChannelId).Find(&mediaProcess).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return err
	}

	//获取软矩阵需要删除的任务
	tasks := make([]live_schedule.CortexTask, 0, 5)
	err = tx.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id=? and media_process_id=? and cortex_output_id=?", c.TenantId, c.ChannelId, mediaProcess.Id, c.CortexOutputId).Find(&tasks).Error
	if err != nil {
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	delScheduleIds := make([]string, 0, 5)
	ids := make([]int, 0, 5)
	offIds := make([]int, 0, 3)
	for _, v := range tasks {
		//停止全球直播
		if len(v.LiveId) > 0 {
			err = liveService.RemoveLive(&liveDto.StopLiveReq{TenantId: v.TenantId, LiveId: v.LiveId})
			if err != nil {
				e.Log.Errorf("StopLive error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}
		//删除协转任务
		if len(v.TaskId) > 0 {
			err = streamHubService.RemoveStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
			if err != nil {
				e.Log.Errorf("StopStreamHub error:%s \r\n", err)
				tx.Rollback()
				return err
			}
			delScheduleIds = append(delScheduleIds, v.TaskId)
		}

		offIds = append(offIds, v.TrafficOffId)
		ids = append(ids, v.Id)
	}
	now := time.Now().Unix()
	//更新调度任务结束时间
	if len(delScheduleIds) > 0 {
		param1 := make(map[string]interface{})
		param1["end_time"] = now
		err := tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id in ? and end_time = 0", delScheduleIds).Updates(param1).Error
		if err != nil {
			e.Log.Errorf("ChannelTaskService updateScheduleRecord error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	offs := make([]live_schedule.TrafficOff, 0, 3)
	err = tx.Model(&live_schedule.TrafficOff{}).Where("id in ?", offIds).Find(&offs).Error
	if err != nil {
		e.Log.Errorf("get TrafficOff error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	offPushes := make(map[int][]dto.PushInfo)
	offPulls := make(map[int][]dto.PullInfo)
	for _, off := range offs {
		pushes := make([]dto.PushInfo, 0, 4)
		pulls := make([]dto.PullInfo, 0, 4)
		if len(off.PushInfos) > 0 {
			json.Unmarshal([]byte(off.PushInfos), &pushes)
			for _, push := range pushes {
				push := push
				if push.CortexOutId == c.CortexOutputId {
					continue
				}
				pushInfos, ok := offPushes[off.Id]
				if !ok {
					pushInfos = make([]dto.PushInfo, 0, 4)
				}
				pushInfos = append(pushInfos, push)
				offPushes[off.Id] = pushInfos
			}
		}

		if len(off.PullInfos) > 0 {
			json.Unmarshal([]byte(off.PullInfos), &pulls)
			for _, pull := range pulls {
				pull := pull
				if pull.CortexOutId == c.CortexOutputId {
					continue
				}
				pullInfos, ok := offPulls[off.Id]
				if !ok {
					pullInfos = make([]dto.PullInfo, 0, 4)
				}
				pullInfos = append(pullInfos, pull)
				offPulls[off.Id] = pullInfos
			}
		}
	}

	for k, off := range offs {
		pushInfos, ok1 := offPushes[off.Id]
		if ok1 {
			push, _ := json.Marshal(&pushInfos)
			*(&offs[k].PushInfos) = string(push)
		}
		pullInfos, ok2 := offPulls[off.Id]
		if ok2 {
			pull, _ := json.Marshal(&pullInfos)
			*(&offs[k].PullInfos) = string(pull)
		}
	}

	//删除软矩阵任务记录
	err = tx.Model(live_schedule.CortexTask{}).Where("id in ?", ids).Delete(&live_schedule.CortexTask{}).Error
	if err != nil {
		e.Log.Errorf("Delete CortexTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//删除输出名记录
	err = tx.Model(live_schedule.CortexOutput{}).Where("id = ?", c.CortexOutputId).Delete(&live_schedule.CortexOutput{}).Error
	if err != nil {
		e.Log.Errorf("Delete CortexOutput error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	if len(offs) > 0 {
		err = tx.Save(&offs).Error
		if err != nil {
			e.Log.Errorf("save off error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	tx.Commit()
	return nil
}

func (e *CortexTaskService) DeleteCortexTaskBySource(c *dto.DeleteCortexTaskReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("Service SaveCortex error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	liveService.Log = e.Log
	streamHubService.Log = e.Log

	//获取矩阵变化
	var mediaProcess live_schedule.MediaProcess
	err := tx.Model(live_schedule.MediaProcess{}).Where("channel_id=? and media_type=2 ", c.ChannelId).Find(&mediaProcess).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return err
	}
	//获取软矩阵需要删除的任务
	tasks := make([]live_schedule.CortexTask, 0, 5)
	err = tx.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id=? and media_process_id=? and cortex_output_id=?", c.TenantId, c.ChannelId, mediaProcess.Id, c.CortexOutputId).Find(&tasks).Error
	if err != nil {
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	ids := make([]int, 0, 5)
	for _, v := range tasks {
		//停止全球直播
		if len(v.LiveId) > 0 {
			err = liveService.RemoveLive(&liveDto.StopLiveReq{TenantId: v.TenantId, LiveId: v.LiveId})
			if err != nil {
				e.Log.Errorf("RemoveLive error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}
		//删除协转任务
		if len(v.TaskId) > 0 {
			err = streamHubService.RemoveStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
			if err != nil {
				e.Log.Errorf("RemoveStreamHub error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

		ids = append(ids, v.Id)
	}

	//删除软矩阵任务记录
	err = tx.Model(live_schedule.CortexTask{}).Where("id in ?", ids).Delete(&live_schedule.CortexTask{}).Error
	if err != nil {
		e.Log.Errorf("Delete CortexTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	tx.Commit()
	return nil
}

func (e *CortexTaskService) SwitchCortexSource(c *dto.SwitchCortexSourceReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("Service SaveCortex error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	liveService.Log = e.Log
	streamHubService.Log = e.Log

	var basic live_schedule.ChannelBasic
	err := tx.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return err
	}

	var cortexOutput live_schedule.CortexOutput
	err = tx.Model(&live_schedule.CortexOutput{}).Where("id=?", c.CortexOutputId).Find(&cortexOutput).Error
	if err != nil {
		e.Log.Errorf("get CortexOutput error:%s \r\n", err)
		return err
	}
	if len(cortexOutput.SwitchTaskId) > 0 && ext.ExtConfig.ServiceConfig.EnableSwitch == 1 {
		err = streamHubService.SwitchSource(&StreamHubDto.ChangeSourceReq{Id: cortexOutput.SwitchTaskId, TenantId: basic.TenantId, Active: c.ActiveItem})
		if err != nil {
			e.Log.Errorf("SwitchSource error:%s \r\n", err)
			return err
		}
	}
	//获取矩阵变化
	var mediaProcess live_schedule.MediaProcess
	err = tx.Model(live_schedule.MediaProcess{}).Where("channel_id=? and media_type=2 ", c.ChannelId).Find(&mediaProcess).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return err
	}

	//获取软矩阵原始任务
	originTasks := make([]live_schedule.CortexTask, 0, 5)
	err = tx.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id=? and media_process_id=? and traffic_on_id=? and cortex_output_id=? ", c.TenantId, c.ChannelId, mediaProcess.Id, c.OriginUpId, c.CortexOutputId).Find(&originTasks).Error
	if err != nil {
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	downIds := make([]int, 0, 5)
	for _, v := range originTasks {
		downIds = append(downIds, v.TrafficOffId)
	}

	//获取所有下车点信息
	trafficOffs := make([]live_schedule.TrafficOff, 0, 5)
	tx.Model(live_schedule.TrafficOff{}).Where("id in ?", downIds).Find(&trafficOffs)
	offMap := make(map[int]live_schedule.TrafficOff)
	for _, v := range trafficOffs {
		v := v
		offMap[v.Id] = v

	}

	//获取软矩阵新目标任务
	targetTasks := make([]live_schedule.CortexTask, 0, 5)
	err = tx.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id=? and media_process_id=? and traffic_on_id=? and cortex_output_id=?", c.TenantId, c.ChannelId, mediaProcess.Id, c.TargetUpId, c.CortexOutputId).Find(&targetTasks).Error
	if err != nil {
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取目标上车点
	var targetTrafficOn live_schedule.TrafficOn
	err = tx.Model(live_schedule.TrafficOn{}).Where("id=?", c.TargetUpId).Find(&targetTrafficOn).Error
	if err != nil {
		e.Log.Errorf("get TrafficOn error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	originMap := make(map[string]struct{})
	for _, v := range originTasks {
		//保存是调度任务的节点
		if v.Area > 0 {
			key := strconv.Itoa(v.CortexOutputId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
			originMap[key] = struct{}{}
		}
	}
	needAddTargets := make([]live_schedule.CortexTask, 0, 5)
	for k, v := range targetTasks {
		v := v
		key := strconv.Itoa(v.CortexOutputId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
		_, ok := originMap[key]
		//原始输出点是自动调度 目标输出点是非自动调度
		if ok && v.Area <= 0 {
			*(&targetTasks[k].Area) = 1
			target := deepcopy.Copy(v).(live_schedule.CortexTask)
			target.Area = 2
			target.Id = 0
			needAddTargets = append(needAddTargets, target)
		}
	}

	targetTasks = append(targetTasks, needAddTargets...)

	//获取目标信源输出地址
	outUdpUrl := ""
	var targetSource live_schedule.SourceBasic
	if targetTrafficOn.StreamType == 1 {
		err = tx.Model(live_schedule.SourceBasic{}).Where("id=?", targetTrafficOn.SourceId).Find(&targetSource).Error
		if err != nil {
			e.Log.Errorf("get targetSource error:%s \r\n", err)
			tx.Rollback()
			return err
		}
		outUdpUrl = targetSource.OutUdpUrl
	} else {
		outUdpUrl = targetTrafficOn.TransUdpUrl
	}
	e.Log.Trace("目标信源：", targetSource.SourceName, outUdpUrl)

	now := time.Now().Unix()
	addCortexTasks := make([]live_schedule.CortexTask, 0, 5)
	delCortexTasks := make([]live_schedule.CortexTask, 0, 5)
	taskIdMap := make(map[string]struct{})
	for _, v := range originTasks {
		v1 := v
		if len(v.TaskId) <= 0 {
			continue
		}
		for _, t := range targetTasks {
			targetTask := t
			//比较同一下车点的任务id
			if (t.CortexOutputId != v.CortexOutputId) || (t.TrafficOffId != v.TrafficOffId) {
				continue
			}
			if v.Area > 0 && t.Area != v.Area {
				continue
			}
			off := offMap[v1.TrafficOffId]
			e.Log.Trace("任务列表：", taskIdMap)
			_, ok := taskIdMap[v1.TaskId]
			if !ok {
				//组建控制协转接口输入参数
				sources := make([]StreamHubDto.Source, 0, 3)

				source := StreamHubDto.Source{}
				source.Url = outUdpUrl
				source.Id = v1.InNodeId
				sources = append(sources, source)

				//最后调用控制协转接口
				req := &StreamHubDto.StreamHubReq{}
				req.Id = v1.TaskId
				req.Name = ""
				req.Type = "live"
				req.TenantId = c.TenantId
				req.Description = ""
				req.Input = StreamHubDto.Input{
					Type:   "push",
					Format: "udp",
					Active: targetSource.ActiveItem,
					Policy: StreamHubDto.Policy{
						Mode: "manual",
					},
					Sources: sources,
				}

				delay := StreamHubDto.Delay{
					Enable:   "no",
					Duration: 10,
				}

				target := StreamHubDto.Target{}
				target.Format = protocols[off.TrafficOffType]
				target.Id = v1.OutNodeId
				target.Url = v1.PushUrl
				target.RedefinePlayUrlEnable = "2"
				req.CallbackAddress = ""
				params := StreamHubDto.RedefinePlayUrlParams{}
				target.RedefinePlayUrlParams = params

				//第三方拉流+内网 输出类型为pull 其他情况为push
				if off.StreamType == 2 && (off.Area == 2 || t.Area == 2) {
					target.Type = "pull"
				} else {
					target.Type = "push"
				}

				if (off.Area == 1 || t.Area == 1) && off.StreamType == 2 {
					target.Format = "rtmp"
				}
				//如果是自定义拉流地址
				if off.StreamType == 2 && off.TrafficOffType == 2 && (off.Area == 2 || off.Area == 3) && off.IsCustom == 1 {
					target.RedefinePlayUrlEnable = "1"
					arr := strings.Split(v1.CustomOutUrl, "/")
					if len(arr) > 1 {
						target.RedefinePlayUrlEnable = "1"
						params.Path = arr[0]
						index := strings.Index(arr[1], ".")
						params.Sub = arr[1][0:index]
						target.RedefinePlayUrlParams = params
					}
				}

				req.Output = StreamHubDto.Output{
					Targets: []StreamHubDto.Target{target},
					Delay:   delay,
					Acl:     StreamHubDto.Acl{},
				}

				//如果不是无缝切换就采用有缝切换
				if len(cortexOutput.SwitchTaskId) <= 0 || ext.ExtConfig.ServiceConfig.EnableSwitch != 1 {
					_, err = streamHubService.UpdateStreamHub(req)
					if err != nil {
						e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
						tx.Rollback()
						return err
					}
				}
				req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/channel-task/cortex-target-call-back"
				taskIdMap[v1.TaskId] = struct{}{}
				targetTask.LiveId = v1.LiveId
				targetTask.CustomOutUrl = v1.CustomOutUrl
				targetTask.TaskId = v1.TaskId
				targetTask.PushUrl = v1.PushUrl
				targetTask.OutUrl = v1.OutUrl
				targetTask.ExtraInfo = v1.ExtraInfo
				targetTask.InUrl = outUdpUrl
				targetTask.InNodeId = v1.InNodeId
				targetTask.OutNodeId = v1.OutNodeId
				targetTask.UpdateTime = now
				targetTask.TaskState = 1
				targetTask.IsEnable = v1.IsEnable
				targetTask.LiveId = v1.LiveId
				targetTask.PushUrl = v1.PushUrl
				targetTask.ExtraInfo = v1.ExtraInfo
				targetTask.StateUpdateTime = now
				targetTask.Area = v1.Area
				e.Log.Trace("添加项目：", targetTask)
				addCortexTasks = append(addCortexTasks, targetTask)
			}
		}
		delTask := v
		////如果是推流地址切换需要把原来的推流地址清空
		//if len(v.PushUrl) > 0 {
		//	delTask.PushUrl = ""
		//}
		//if len(v.CustomOutUrl) > 0 {
		//	delTask.CustomOutUrl = ""
		//}
		delTask.TaskId = ""
		delTask.LiveId = ""
		delTask.TaskState = 3
		delTask.IsEnable = 2
		delCortexTasks = append(delCortexTasks, delTask)
	}
	e.Log.Trace("删除的任务：", delCortexTasks)
	if len(addCortexTasks) > 0 {
		e.Log.Trace("添加的任务：", addCortexTasks)
		err = tx.Model(&addCortexTasks).Save(&addCortexTasks).Error
		if err != nil {
			e.Log.Errorf("Save addCortexTasks error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	if len(delCortexTasks) > 0 {
		err = tx.Model(&delCortexTasks).Save(&delCortexTasks).Error
		if err != nil {
			e.Log.Errorf("Save delCortexTasks error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	tx.Commit()
	return nil
}

func (e *CortexTaskService) GetCortexTaskList(c *dto.GetCortexTaskListReq) (*dto.CortexTaskRes, error) {

	res := dto.CortexTaskRes{}

	cortexTaskModels := make([]live_schedule.CortexTaskModel, 0, 5)
	//获取通道详情
	var basic live_schedule.ChannelBasic
	err := e.Orm.Model(live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("GetChannelBasic error:%s \r\n", err)
		return nil, err
	}

	now := time.Now().Unix()
	//获取矩阵变化
	var mediaProcess live_schedule.MediaProcess
	err = e.Orm.Model(live_schedule.MediaProcess{}).Where("channel_id=? and media_type=2 ", c.ChannelId).Find(&mediaProcess).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return nil, err
	}

	//获取棋盘格子任务
	tasks := make([]live_schedule.CortexTask, 0, 5)
	db := e.Orm.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id=? and media_process_id=?", c.TenantId, c.ChannelId, mediaProcess.Id)
	if (len(c.CortexOutputIds) > 0 && len(c.SourceIds) > 0) || (len(c.CortexOutputIds) <= 0 && len(c.OutPutName) > 0) || (len(c.SourceIds) <= 0 && len(c.InputName) > 0) {
		db.Where("cortex_output_id in ? and source_id in ?", c.CortexOutputIds, c.SourceIds)
	} else if len(c.OutPutName) > 0 || len(c.InputName) > 0 {
		db.Where("cortex_output_id in ? or source_id in ?", c.CortexOutputIds, c.SourceIds)
	}

	err = db.Find(&tasks).Error
	if err != nil {
		e.Log.Errorf("GetCortexTask error:%s \r\n", err)
		return nil, err
	}

	res.CortexTaskModels = &cortexTaskModels
	//获取输出名列表
	cortexOutputs := make([]live_schedule.CortexOutput, 0, 5)
	db1 := e.Orm.Model(live_schedule.CortexOutput{}).Where("tenant_id=? and channel_id=? and media_process_id=?", c.TenantId, c.ChannelId, mediaProcess.Id)
	if len(c.CortexOutputIds) > 0 {
		db1.Where("id in ?", c.CortexOutputIds)
	}
	err = db1.Find(&cortexOutputs).Error
	if err != nil {
		e.Log.Errorf("get CortexOutput error:%s \r\n", err)
		return nil, err
	}
	if len(c.CortexOutputIds) <= 0 && len(c.OutPutName) > 0 {
		cortexOutputs = make([]live_schedule.CortexOutput, 0, 5)
	}
	res.CortexOutputs = &cortexOutputs

	//获取上车点和媒体处理关系
	upIds := make([]int, 0, 5)
	err = e.Orm.Model(live_schedule.UpMedia{}).Select("up_id").Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&upIds).Error
	if err != nil {
		e.Log.Errorf("get UpMedia error:%s \r\n", err)
		return nil, err
	}

	if len(c.SourceIds) <= 0 && len(c.InputName) > 0 || (len(c.InputName) > 0 && len(c.OutPutName) > 0 && len(cortexOutputs) <= 0) {
		upIds = make([]int, 0, 5)
	}

	//单信源上车点
	upModels := make([]live_schedule.TrafficOnModel, 0, 5)
	db2 := e.Orm.Model(live_schedule.TrafficOn{}).
		Select("traffic_on.*,source_basic.source_name").
		Joins(" left join source_basic on source_basic.id=traffic_on.source_id").
		Where("traffic_on.id in ?", upIds)

	if len(c.SourceIds) > 0 {
		db2.Where("source_id in ?", c.SourceIds)
	}
	err = db2.Find(&upModels).Error
	if err != nil {
		e.Log.Errorf("get TrafficOn error:%s \r\n", err)
		return nil, err
	}

	//获取上车点
	originSourceIds := make([]int, 0, 5)
	transSourceIds := make([]int, 0, 5)
	for _, v := range upModels {
		if v.StreamType == 1 {
			originSourceIds = append(originSourceIds, v.SourceId)
		} else {
			originSourceIds = append(originSourceIds, v.SourceId)
			transSourceIds = append(transSourceIds, v.SourceId)
		}
	}

	//获取信源基本详情
	basicService := &SourceBasicService{}
	basicService.Orm = e.Orm
	basicService.Log = e.Log
	originSourceMap := make(map[int]dto.SourceModelRes)
	transSourceMap := make(map[int][]live_schedule.SourceTemplateModel)
	var count int64
	pagination := commonDto.Pagination{PageIndex: 1, PageSize: 1000}

	//获取原始流信息
	if len(originSourceIds) > 0 {
		basics, err := basicService.GetPage(&dto.SourceBasicGetPageReq{Pagination: pagination, SourceIds: originSourceIds, IsDelete: 2, Kind: 3}, &count)
		if err != nil {
			e.Log.Errorf("ChannelTaskService GetSourceBasic error:%s \r\n", err)
			return nil, err
		}
		for _, v := range *basics {
			v := v
			originSourceMap[v.Basic.Id] = v
		}
	}

	//获取转码流信息
	if len(transSourceIds) > 0 {
		trans, err := basicService.SourcesTransCodeTaskPage(&dto.GetSourceTaskPageReq{Pagination: pagination, SourceIds: transSourceIds}, &count)
		if err != nil {
			e.Log.Errorf("get SourcesTransCodeTaskPage error:%s \r\n", err)
			return nil, err
		}
		for _, v := range *trans {
			v := v
			transSources, ok := transSourceMap[v.SourceTemplate.SourceId]
			if !ok {
				transSources = make([]live_schedule.SourceTemplateModel, 0, 5)
			}
			transSources = append(transSources, v)
			transSourceMap[v.SourceTemplate.SourceId] = transSources
		}
	}

	trafficOnRes := make([]dto.TrafficOnRes, 0, 5)
	for _, v := range upModels {
		v := v
		if v.IsGroup == 1 {
			continue
		}
		t := dto.TrafficOnRes{}
		t.TrafficOn = &v
		if v.StreamType == 1 {
			//信源已被删除
			basicRes, ok := originSourceMap[v.SourceId]
			if !ok {
				continue
			}
			if basicRes.Basic.ProtocolType == 5 && basicRes.Basic.SourceState != 2 {
				t.SourceState = 1
			} else {
				t.SourceState = basicRes.Basic.SourceState
			}

			t.SourceModelRes = &basicRes
			t.TrafficOn.SourceName = basicRes.Basic.SourceName

		} else {
			//转码任务已被删除
			tans, ok := transSourceMap[v.SourceId]
			if !ok {
				continue
			}
			for _, tran := range tans {
				for _, s := range tran.TransCodeTasks {
					if v.TransRecordId == s.TransCodeTask.TaskRecordId {
						//转码任务停用
						if tran.SourceTemplate.IsEnable == 2 {
							t.SourceState = 2
							break
						}
						//转码任务异常
						if s.TransCodeTask.NodeState == 2 {
							t.SourceState = 3
							break
						}
						t.SourceState = 1
					}
				}
			}
			t.SourceTemplateModels = &tans
		}
		trafficOnRes = append(trafficOnRes, t)
	}

	res.TrafficOns = &trafficOnRes

	//获取媒体处理下车点关系
	mediaOffs := make([]int, 0, 5)
	err = e.Orm.Model(live_schedule.MediaOff{}).Select("off_id").Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&mediaOffs).Error
	if err != nil {
		e.Log.Errorf("get MediaOff error:%s \r\n", err)
		return nil, err
	}

	//获取车点关
	trafficOffs := make([]live_schedule.TrafficOff, 0, 5)
	err = e.Orm.Model(live_schedule.TrafficOff{}).Where("id in ?", mediaOffs).Find(&trafficOffs).Error
	if err != nil {
		e.Log.Errorf("get TrafficOff error:%s \r\n", err)
		return nil, err
	}
	res.TrafficOffs = &trafficOffs

	taskMap := make(map[string][]live_schedule.CortexTask)
	for _, v := range tasks {
		v := v
		key := strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.CortexOutputId)
		tempTasks, ok := taskMap[key]
		if !ok {
			tempTasks = make([]live_schedule.CortexTask, 0, 5)
		}
		tempTasks = append(tempTasks, v)
		taskMap[key] = tempTasks
	}
	for k, v := range taskMap {
		v := v
		model := live_schedule.CortexTaskModel{}
		//获取上车点和输出点
		keys := strings.Split(k, models.Delimiter)
		if len(keys) > 1 {
			upId, _ := strconv.Atoi(keys[0])
			outputId, _ := strconv.Atoi(keys[1])
			model.UpId = upId
			model.CortexOutputId = outputId
		}
		//获取棋盘格子状态
		for _, t := range v {
			if len(t.TaskId) <= 0 {
				continue
			}
			state := e.GetCortexState(t, basic, originSourceMap, now)
			model.CortexState = state
		}
		e.Log.Trace("软矩阵任务状态：", k, model.CortexState)
		model.CortexTasks = v
		cortexTaskModels = append(cortexTaskModels, model)
	}

	return &res, nil

}

func (e *CortexTaskService) GetCortexState(c live_schedule.CortexTask, basic live_schedule.ChannelBasic, sourceMap map[int]dto.SourceModelRes, now int64) int {

	state := 0
	source := sourceMap[c.SourceId]

	// 1 原色 2 灰色 3 绿色 4 红色 5切换中
	//1 绿色  2 红色 3 原色 4 切换 5 灰色

	//信源和调度任务任意停用且都不为异常
	if basic.ChannelState == 2 || source.Basic.SourceState == 2 && !(basic.ChannelState == 3 && source.Basic.SourceState == 3) {
		if c.IsEnable == 1 {
			state = 5
		} else {
			state = 3
		}
	}

	//信源和调度任务任意启用且都不为异常
	if basic.ChannelState == 1 && source.Basic.SourceState == 1 && !(basic.ChannelState == 3 && source.Basic.SourceState == 3) {
		if c.IsEnable == 1 {
			if c.StateUpdateTime+10 < now {
				if c.TaskState == 1 {
					state = 1
				}
				if c.TaskState == 2 && len(c.TaskId) > 0 {
					state = 2
				}
			} else {
				state = 4
			}
		} else {
			state = 3
		}
	}

	//信源和调度任务任意为异常
	if basic.ChannelState == 3 || source.Basic.SourceState == 3 {
		if c.IsEnable == 1 {
			state = 2
		} else {
			state = 3
		}
	}
	e.Log.Trace("当前矩阵状态：", state, c.Id)
	return state
}

func (e *CortexTaskService) StopCortexTask(c *dto.StopCortexTaskReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("Service CreateCortexTask error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		return err
	}
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	liveService.Log = e.Log
	streamHubService.Log = e.Log
	//now := time.Now().Unix()

	//获取矩阵变化
	var mediaProcess live_schedule.MediaProcess
	err := tx.Model(live_schedule.MediaProcess{}).Where("channel_id=? and media_type=2 ", c.ChannelId).Find(&mediaProcess).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return err
	}

	//获取原始任务
	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	err = tx.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id =? and traffic_on_id=? and cortex_output_id=? and media_process_id=?", c.TenantId, c.ChannelId, c.UpId, c.CortexOutputId, mediaProcess.Id).Find(&cortexTasks).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		return err
	}

	//停止矩阵任务
	e.Log.Trace("查询的任务数量：", len(cortexTasks))
	stopIds := make([]int, 0, 5)
	for _, v := range cortexTasks {
		v := v
		if len(v.TaskId) > 0 {
			err = streamHubService.RemoveStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
			if err != nil {
				tx.Rollback()
				e.Log.Errorf("RemoveStreamHub error:%s \r\n", err)
				return err
			}
			stopIds = append(stopIds, v.Id)
		}
	}

	if len(stopIds) > 0 {
		param := make(map[string]interface{})
		param["is_enable"] = 2
		param["task_id"] = ""
		param["task_state"] = 3
		param["update_time"] = time.Now().Unix()
		param["state_update_time"] = time.Now().Unix()
		err = tx.Model(live_schedule.CortexTask{}).Where("id in ?", stopIds).Updates(param).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("update CortexTask error:%s \r\n", err)
			return err
		}
	}

	tx.Commit()
	return nil
}

func (e *CortexTaskService) StartCortexTask(c *dto.StartCortexTaskReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("Service CreateCortexTask error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		return err
	}
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	liveService.Log = e.Log
	streamHubService.Log = e.Log
	//now := time.Now().Unix()

	//获取通道任务
	var basic live_schedule.ChannelBasic
	err := tx.Model(live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		return err
	}

	//获取矩阵变化
	var mediaProcess live_schedule.MediaProcess
	err = tx.Model(live_schedule.MediaProcess{}).Where("channel_id=? and media_type=2 ", c.ChannelId).Find(&mediaProcess).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return err
	}

	//获取原始任务
	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	err = tx.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id =? and traffic_on_id=? and cortex_output_id=? and media_process_id=?", c.TenantId, c.ChannelId, c.UpId, c.CortexOutputId, mediaProcess.Id).Find(&cortexTasks).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		return err
	}

	ids := make([]int, 0, 5)
	for _, v := range cortexTasks {
		if len(v.TaskId) > 0 {
			ids = append(ids, v.Id)
		}
	}
	if basic.ChannelState == 2 {
		param := make(map[string]interface{})
		param["is_enable"] = 1
		param["update_time"] = time.Now().Unix()
		param["state_update_time"] = time.Now().Unix()
		err = e.Orm.Model(live_schedule.CortexTask{}).Where("id in ?", ids).Updates(param).Error
		if err != nil {
			e.Log.Errorf("update CortexTask error:%s \r\n", err)
			return err
		}
		tx.Commit()
		return nil
	}

	//开始矩阵任务
	for _, v := range cortexTasks {
		v := v
		if len(v.TaskId) > 0 {
			err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
			if err != nil {
				tx.Rollback()
				e.Log.Errorf("StartStreamHub error:%s \r\n", err)
				return err
			}
		}
	}

	param := make(map[string]interface{})
	param["is_enable"] = 1
	param["update_time"] = time.Now().Unix()
	param["state_update_time"] = time.Now().Unix()
	err = e.Orm.Model(live_schedule.CortexTask{}).Where("id in ?", ids).Updates(param).Error
	if err != nil {
		e.Log.Errorf("update CortexTask error:%s \r\n", err)
		return err
	}

	tx.Commit()
	return nil
}

func (e *CortexTaskService) GetCortexOutInfo(tx *gorm.DB, c *dto.GetCortexOutInfo) (*live_schedule.CortexTaskModel, error) {
	res := live_schedule.CortexTaskModel{}
	cortexOutInfos := make([]live_schedule.CortexTask, 0, 5)
	err := e.Orm.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id=? and cortex_output_id=? and (task_id !='' or task_id !=null)", c.TenantId, c.ChannelId, c.CortexOutputId).Find(&cortexOutInfos).Error
	cortexOutAllInfos := make([]live_schedule.CortexTask, 0, 5)
	err = e.Orm.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id=? and cortex_output_id=?", c.TenantId, c.ChannelId, c.CortexOutputId).Find(&cortexOutAllInfos).Error
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("get CortexOutput error:%s \r\n", err)
		return nil, err
	}

	now := time.Now().Unix()
	//获取通道详情
	var basic live_schedule.ChannelBasic
	err = e.Orm.Model(live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Find(&basic).Error
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("GetChannelBasic error:%s \r\n", err)
		return nil, err
	}
	//获取上车点
	upIds := make([]int, 0, 5)
	err = e.Orm.Model(live_schedule.UpMedia{}).Select("up_id").Where("channel_id=?", c.ChannelId).Find(&upIds).Error
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("GetUpMedia error:%s \r\n", err)
		return nil, err
	}

	ups := make([]live_schedule.TrafficOn, 0, 5)
	err = e.Orm.Model(live_schedule.TrafficOn{}).Where("id in ?", upIds).Find(&ups).Error
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("GetTrafficOn error:%s \r\n", err)
		return nil, err
	}

	upMap := make(map[int]live_schedule.TrafficOn)
	for _, v := range ups {
		v := v
		upMap[v.Id] = v
	}

	//获取下车点
	offIds := make([]int, 0, 5)
	err = e.Orm.Model(live_schedule.MediaOff{}).Select("off_id").Where("channel_id=?", c.ChannelId).Find(&offIds).Error
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("GetMediaOff error:%s \r\n", err)
		return nil, err
	}

	offs := make([]live_schedule.TrafficOff, 0, 5)
	err = e.Orm.Model(live_schedule.TrafficOff{}).Where("id in ?", offIds).Find(&offs).Error
	if err != nil {
		//tx.Rollback()
		e.Log.Errorf("GetTrafficOff error:%s \r\n", err)
		return nil, err
	}

	offMap := make(map[int]live_schedule.TrafficOff)
	for _, v := range offs {
		v := v
		offMap[v.Id] = v
	}

	flagMap := make(map[string]struct{})
	for _, v := range cortexOutAllInfos {
		key := strconv.Itoa(v.TrafficOffId) + models.Delimiter + strconv.Itoa(v.CortexOutputId)
		_, ok := flagMap[key]
		if ok {
			continue
		}
		if len(v.TaskId) > 0 {
			flagMap[key] = struct{}{}
		}
	}

	//查询转推或者自定义地址
	outInfos := make([]live_schedule.CortexInfo, 0, 5)
	offUrlMap := make(map[string]struct{})
	for _, v := range cortexOutAllInfos {

		up := upMap[v.TrafficOnId]
		off := offMap[v.TrafficOffId]

		//信源或者转码删除不显示地址
		if up.IsDelete == 1 {
			continue
		}

		//如果下车点不是转推或者自定义地址 并且没有输出任务不显示地址
		if len(v.TaskId) <= 0 && !(off.StreamType == 1 || off.IsCustom == 1) {
			continue
		}

		if off.StreamType == 1 && len(v.PushUrl) <= 0 {
			continue
		}

		if off.IsCustom == 1 && len(v.CustomOutUrl) <= 0 {
			continue
		}

		if c.UpId > 0 && v.TrafficOnId != c.UpId {
			continue
		}

		key := strconv.Itoa(v.TrafficOffId) + models.Delimiter + strconv.Itoa(v.CortexOutputId)
		_, ok1 := flagMap[key]
		_, ok := offUrlMap[key]
		if (!ok && len(v.TaskId) > 0) || (!ok && !ok1) {
			info := live_schedule.CortexInfo{}
			info.OffId = v.TrafficOffId
			info.PullUrl = v.CustomOutUrl
			info.PushUrl = v.PushUrl
			info.OutUrl = v.OutUrl
			info.CortexOutId = v.CortexOutputId
			outInfos = append(outInfos, info)
			offUrlMap[key] = struct{}{}
			e.Log.Trace("AAAAAAAAAAAAAA:", v.PushUrl, v.Id)
		}
	}

	for _, v := range cortexOutInfos {
		if v.IsEnable == 1 {
			//异常
			if v.TaskState == 2 {
				res.CortexState = 2
				break
			} else {
				//正常
				if v.StateUpdateTime+10 < now {
					if v.TaskState == 1 {
						res.CortexState = 1
					}
				} else {
					res.CortexState = 4
				}
			}
		} else {
			//离线
			res.CortexState = 3
		}
	}
	if len(cortexOutInfos) > 0 {
		res.UpId = cortexOutInfos[0].TrafficOnId
	}
	res.CortexOutputId = c.CortexOutputId
	res.CortexTasks = cortexOutInfos
	res.CortexInfos = outInfos
	return &res, nil
}

func (e *CortexTaskService) GetCortexByFilter(c *dto.GetCortexReq) (*dto.CortexTaskRes, error) {

	if len(c.InputName) <= 0 && len(c.OutPutName) <= 0 {
		return nil, nil
	}

	outs := make([]int, 0, 5)
	if len(c.OutPutName) > 0 {
		//获取输出地址
		err := e.Orm.Model(&live_schedule.CortexOutput{}).Select("id").Where("channel_id=? and out_name like ?", c.ChannelId, "%"+c.OutPutName+"%").Find(&outs).Error
		if err != nil {
			e.Log.Errorf("get CortexOutput error:%s \r\n", err)
			return nil, err
		}
	}

	sources := make([]int, 0, 5)
	if len(c.InputName) > 0 {
		//获取输入地址
		sourceIds := make([]string, 0, 5)
		err := e.Orm.Model(&live_schedule.TrafficOn{}).Select("source_id").Where("channel_id=?", c.ChannelId).Find(&sourceIds).Error
		if err != nil {
			e.Log.Errorf("get TrafficOn error:%s \r\n", err)
			return nil, err
		}

		err = e.Orm.Model(&live_schedule.SourceBasic{}).Select("id").Where("id in ? and source_name like ?", sourceIds, "%"+c.InputName+"%").Find(&sources).Error
		if err != nil {
			e.Log.Errorf("get SourceBasic error:%s \r\n", err)
			return nil, err
		}
	}

	res, err := e.GetCortexTaskList(&dto.GetCortexTaskListReq{TenantId: c.TenantId, OutPutName: c.OutPutName, InputName: c.InputName, ChannelId: c.ChannelId, CortexOutputIds: outs, SourceIds: sources})
	if err != nil {
		e.Log.Errorf("GetCortexTaskList error:%s \r\n", err)
		return nil, err
	}
	return res, nil
}

func (e *CortexTaskService) GenerateInsertLiveParam(tenantId string) *liveDto.LiveStreamInsertReq {
	streamReq := &liveDto.LiveStreamInsertReq{}
	streamReq.EventName = ""
	streamReq.TenantId = tenantId
	streamReq.ExpireTime = 10 * 365 * 24 * 60 * 60
	streamReq.IMEnable = 0
	streamReq.WebPusherEnable = 0
	streamReq.PullOrPush = ""
	streamReq.AutoRecord = "0"
	streamReq.EventType = "1"
	streamReq.ServiceArea = "cn_mainland"
	return streamReq
}

func (e *CortexTaskService) CortexTargetCallBack(c *dto.TargetCallBackReq) error {

	//更新通道状态
	param := make(map[string]interface{})
	if c.TransferState == "Capturing" {
		param["task_state"] = 1
	} else {
		param["task_state"] = 2
	}
	param["state_update_time"] = c.TransferStateLastUpdateTime
	err := e.Orm.Model(live_schedule.CortexTask{}).Where("task_id=? and out_node_id=?", c.Id, c.TaskId).Updates(param).Error
	if err != nil {
		e.Log.Errorf("CortexTargetCallBack error:%s \r\n", err)
		return err
	}
	return nil
}

func (e *CortexTaskService) CheckUpUsed(c *dto.CheckUpUsedReq) (int, error) {

	results := make([]live_schedule.CortexTask, 0, 5)
	err := e.Orm.Model(&live_schedule.CortexTask{}).Where("traffic_on_id =? and is_enable=1", c.UpId).Find(&results).Error
	if err != nil {
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		return 0, err
	}
	if len(results) > 0 {
		return 1, nil
	}
	return 0, nil
}

func (e *CortexTaskService) CalculateProgress(partNum int, workTotalNum int, basic live_schedule.ChannelBasic) error {
	result := float64(partNum) / float64(workTotalNum)
	formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
	e.Log.Trace("当前进度为", formattedResult)
	param := make(map[string]interface{})
	param["progress"] = formattedResult
	if workTotalNum <= partNum {
		param["channel_state"] = basic.ChannelState
	}
	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Updates(&param).Error
	if err != nil {
		e.Log.Trace("update ChannelBasic error：", err)
	}
	return err
}

func (e *CortexTaskService) UpdateCortexSwitchTask(workTotalNum int, partNum *int, tasks []live_schedule.CortexTask, addTasks []live_schedule.CortexTask, outputs []live_schedule.CortexOutput, transferUpPointIds []int, upMap map[int]live_schedule.TrafficOn, sourcesMap map[int]live_schedule.SourceBasic, basic live_schedule.ChannelBasic, streamHubService *streamHubService.StreamHubService, liveBasicService SourceBasicService) error {

	//var m sync.Mutex
	liveBasicService.Orm = e.Orm
	// 创建任务通道
	taskCh := make(chan live_schedule.CortexOutput, 3)
	go func() {
		for i := 0; i < len(outputs); i++ {
			taskCh <- outputs[i] // 将任务放入通道
		}
		close(taskCh) // 关闭通道，表示任务已经全部放入通道
	}()

	for i := 0; i < 3; i++ {
		go func() {
			for v := range taskCh {
				v := v
				//如果是软矩阵并且支持vdcm则需要将信源加入vdcm
				e.Log.Trace("进入切换模式")
				switchReq := &StreamHubDto.SwitchSourceReq{}
				switchReq.Id = v.SwitchTaskId
				switchReq.Name = ""
				switchReq.TenantId = v.TenantId
				switchInput := StreamHubDto.SwitchInput{}
				switchInput.Type = "push"
				switchInput.Format = "udp"
				switchInput.Active = 0
				switchInput.Policy = StreamHubDto.Policy{Mode: "manual"}
				switchInput.Sources = make([]StreamHubDto.Source, 0, 3)
				switchReq.Type = "live"

				//原始流切换
				for _, u := range transferUpPointIds {
					up := upMap[u]
					switchSource := StreamHubDto.Source{}
					source := sourcesMap[up.SourceId]
					if up.StreamType == 1 {
						switchSource.Url = source.OutUdpUrl
					} else {
						switchSource.Url = up.TransUdpUrl
					}

					for _, t := range tasks {
						if t.TrafficOnId == u {
							switchSource.Id = t.SwitchNodeId
						}
					}
					switchInput.Sources = append(switchInput.Sources, switchSource)
				}
				switchReq.SwitchInput = switchInput
				switchRes := &StreamHubDto.SwitchSourceRes{}
				switchRes, err := streamHubService.UpdateSwitchSourceTask(switchReq)
				if err != nil {
					e.Log.Errorf("UpdateSwitchSourceTask error:%s \r\n", err)
				}
				if switchRes != nil {
					params := make(map[string]interface{})
					params["switch_task_id"] = switchRes.TaskId
					params["out_udp_url"] = switchRes.OutputAddress
					err = e.Orm.Model(live_schedule.CortexOutput{}).Where("id=?", v.Id).Updates(&params).Error
					if err != nil {
						e.Log.Errorf("update MediaProcess error:%s \r\n", err)
					}
					//保存切换任务的主备辅id
					for _, v := range switchRes.Input.Sources {
						v := v
						for k, s := range tasks {
							source := sourcesMap[s.SourceId]
							up := upMap[s.TrafficOnId]
							if (v.Url == source.OutUdpUrl && up.StreamType == 1) || (v.Url == source.TransOutUdpUrl && up.StreamType == 2) {
								*(&tasks[k].SwitchNodeId) = v.Id
							}
						}

						for k, s := range addTasks {
							source := sourcesMap[s.SourceId]
							up := upMap[s.TrafficOnId]
							if (v.Url == source.OutUdpUrl && up.StreamType == 1) || (v.Url == source.TransOutUdpUrl && up.StreamType == 2) {
								e.Log.Trace("KKKKKKKKKKKKKKKK:", source.OutUdpUrl, up.StreamType)
								*(&addTasks[k].SwitchNodeId) = v.Id
							}
						}
					}
					if len(tasks) > 0 {
						err = e.Orm.Save(&tasks).Error
						if err != nil {
							e.Log.Errorf("save cortexTasks error:%s \r\n", err)
						}
					}
				}
				e.Log.Trace("获取的原始流切换地址：", switchRes.OutputAddress)
				//m.Lock()
				//*partNum++
				//err = e.CalculateProgress(*partNum, workTotalNum, basic)
				//if err != nil {
				//	e.Log.Trace("CalculateProgress error：", err)
				//}
				//m.Unlock()
			}
		}()
	}

	return nil

}
